package day1


import org.apache.log4j.{Level, Logger}
import org.apache.spark.{SparkConf, SparkContext}

object RDD {
  def main(args: Array[String]): Unit = {
    //定义hadoop内容
    System.setProperty("hadoop.home.dir", "D:\\hadoop")
    //    //配置spark
    //    val conf = new SparkConf()
    //    //任务名称
    //    conf.setAppName("Basic")
    /*
      设置spark工作模式
      local:本地工作模式,更使用学习spark的操作处理,更方便调用,只适合学习使用
      Standalone: 集群化环境
      Yarn: Hadoop集群
     */
    //    conf.setMaster("local")

    //基于scala中一切皆为对象，可以配置Spark操作合并
    //    new SparkConf().setAppName("Basic").setMaster("local")

    //关闭日志
    Logger.getLogger("org").setLevel(Level.OFF)

    //创建一个spark的任务入口:SparkC

    val sc = new SparkContext(new SparkConf().setAppName("Basic").setMaster("local"))

    //RDD的数据内容来自两部分
    //1.读取来自外部的数据集内容
    val rdd_file = sc.textFile("D:\\data\\HCIP\\files\\wordCpunt.txt")
    //2.在驱动程序中进行数据获取
    val arr = Array(1, 2, 3, 4, 5, 6, 7, 8, 9)
    val rdd = sc.parallelize(arr)

    /*
      transformation  action
      RDD 有两种算子,转换算子，行动算子
      转换算子: 由一个RDD生成另一个RDD，转换操作针对于前一个RDD中的各个元素
      行动算子: 计算各个算子转换操作
      转换算子生成的内容一定是RDD类型
      惰性计算?
      因为spark是基于内存进行处理操作的,为了节约内存空间，避免别浪费
      所以才会在遇到行动算子才会运行
     */


    //使用一个转换算子map生成一个新的RDD类型:rdd_map
    //转换算子map:读取钱一个RDD的数据，通过计算防止在新的RDD上（前一个RDD、新RDD长度一致）
    val rdd_map = rdd.map(i => println("当前的数值的乘积为:" + i * i))
    arr.foreach(i => println(i * i))
    //    rdd_map.foreach(println)
    //执行算子 count()
    //rdd_map.count()
    //执行算子:first:读取RDD的第一个数据
    //对于spark来说，当了解了完整的操作链，计算只需要用到的数据
    rdd_map.first()

    val arr1 = Array(1, 2, 3, 4, 5, 6, 7, 8, 9)
    val arr2 = Array(1, 2, 3, 4, 5, 6, 7, 8, 9)
    val rdd1 = sc.parallelize(arr1)
    val rdd2 = sc.parallelize(arr2)

    //转换算子: union() 将前一个RDD的数据于后一个RDD的数据合并,生成一个新的RDD
    val rdd1_map = rdd1.map(i => println(s"RDD1:$i"))
    val rdd2_map = rdd2.map(i => println(s"RDD1:$i"))

    val rddADD = rdd1.union(rdd2)

    rddADD.foreach(println)
    //通过谱系图例子，可以得到一个谱系图
    //谱系图的作用：通过表达依赖关系从而可以在出现父RDD丢失的情况下进行数据恢复
    rdd1_map.foreach(println)
    rdd2_map.foreach(println)
    rdd1_map.count()

  }

}
